package com.fudan.run.trans;

import java.util.Map;

import org.apache.spark.api.java.JavaRDD;

import com.fudan.cfg.base.TransDef;
import com.fudan.cfg.trans.TestNode;
import com.fudan.run.JobRunner;
import com.fudan.run.VariableRddMap;

import scala.Tuple2;

public class TestTrans extends TransHandler<TestNode>{

	@Override
	public void trans(JobRunner jobRunner,TestNode testNode, VariableRddMap variableMap) {
		JavaRDD<Map<String, Object>> target = variableMap.getRdd(testNode.getTarget());

		JavaRDD<Map<String, Object>> after = target.mapToPair(l->Tuple2.apply(l.get("bf_StudentID"), l))
		.leftOuterJoin(variableMap.getRdd("chengji2").mapToPair(l->Tuple2.apply(l.get("mes_StudentID"), l)))
		.map(l->{
			Map<String,Object> map = l._2._1();
			Map<String, Object> child = l._2._2().orNull();
			if(child!=null) {
				map.put("chengji", child);
			}
			return map;
		});
		variableMap.addRdd(testNode.getName(), after);
	}

}
